package com.liuqi.knet.remoting.transport.netty;

import com.liuqi.knet.Constants;
import com.liuqi.knet.remoting.Channel;
import com.liuqi.knet.remoting.ChannelHandler;
import com.liuqi.knet.remoting.ClientConfig;
import com.liuqi.knet.remoting.RemotingException;
import com.liuqi.knet.remoting.transport.AbstractClient;
import com.liuqi.knet.remoting.transport.dispatcher.DispatcherHandlers;
import com.liuqi.knet.remoting.transport.netty.logging.NettyHelper;
import com.liuqi.loggers.api.Logger;
import com.liuqi.loggers.facade.LoggersFacade;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * @author liuqi
 * @date 2025/3/6 14:02
 **/
public class NettyClient extends AbstractClient {

    static {
        NettyHelper.setNettyLoggerFactory();
    }

    private static final Logger logger = LoggersFacade.getLogger(NettyClient.class);

    private Bootstrap bootstrap;

    private EventLoopGroup workerGroup;

    private volatile io.netty.channel.Channel channel;

    public NettyClient(ChannelHandler handler, ClientConfig config) throws RemotingException {
        super(new DispatcherHandlers(handler, config).warpClient(), config);
    }

    @Override
    protected void doOpen() throws Throwable {
        final NettyClientHandler nettyClientHandler = new NettyClientHandler(getChannelHandler(), getNetConfig());
        bootstrap = new Bootstrap();

        workerGroup = new NioEventLoopGroup(Constants.workerThreads,
                new DefaultThreadFactory(Constants.clientWorkerGroupName, true));

        bootstrap.group(workerGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .channel(NioSocketChannel.class);

        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());

        bootstrap.handler(new ChannelInitializer() {
            @Override
            protected void initChannel(io.netty.channel.Channel ch) throws Exception {
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec());
                ch.pipeline()
                        .addLast("decoder", adapter.getDecoder())
                        .addLast("encoder", adapter.getEncoder())
                        .addLast("handler", nettyClientHandler);
            }
        });

        logger.info("client opened");
    }

    @Override
    protected void doConnect() throws Throwable {
        logger.info("发起客户端连接, " + getRemoteAddress());

        ChannelFuture future = bootstrap.connect(getRemoteAddress());
        boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
        if (ret && future.isSuccess()) {
            // 连接成功
            io.netty.channel.Channel newChannel = future.channel();

            // 关闭旧的通道
            io.netty.channel.Channel oldChannel = NettyClient.this.channel;
            try {
                if (oldChannel != null) {
                    try {
                        logger.info("关闭旧的连接: " + oldChannel + ", 并创建新的连接: " + newChannel);
                        oldChannel.close();
                    } finally {
                        NettyChannel.removeChannelIfDisconnected(oldChannel);
                    }
                }
            } finally {
                if (NettyClient.this.isClosed()) {
                    try {
                        logger.info("客户端已经停掉了, 准备关闭新的连接.");
                        newChannel.close();
                    } finally {
                        NettyClient.this.channel = null;
                        NettyChannel.removeChannelIfDisconnected(newChannel);
                    }
                } else {
                    NettyClient.this.channel = newChannel;
                }
            }
        } else if (future.cause() != null) {
            // 发生错误
            throw new RemotingException("客户端无法连接远程服务端. 远程服务端: " + getRemoteAddress()
                    + ", 错误信息: " + future.cause().getMessage(), future.cause());
        } else {
            // 超时
            throw new RemotingException("客户端无法连接远程服务端. 远程服务端: " + getRemoteAddress() + ", " +
                    "客户端连接超时: 超时时间[" + getConnectTimeout() + "]ms");
        }
    }

    @Override
    protected void doClose() throws Throwable {

    }

    @Override
    protected void doDisConnect() throws Throwable {
        try {
            NettyChannel.removeChannelIfDisconnected(channel);
        } catch (Throwable t) {
            logger.warn(t.getMessage());
        }
    }

    @Override
    protected Channel getChannel() {
        io.netty.channel.Channel c = channel;
        if (c == null || !c.isActive())
            return null;
        return NettyChannel.getOrAddChannel(c, getChannelHandler(), getNetConfig());
    }

    @Override
    public String toString() {
        return "NettyClient{" +
                "remoteAddress=" + getRemoteAddress() +
                '}';
    }
}
